package com.itcast.zk.locker.service;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.curator.shaded.com.google.common.collect.Ordering;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * <p>Description: 共享锁 </p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
@Service("sharedLocker")
public class SharedLocker implements InitializingBean, ILocker {
    
    private static final Logger LOGGER = LoggerFactory.getLogger(SharedLocker.class);
    
    public static final String OPT_WRITE = "W";
    public static final String OPT_READ = "R";
    
    private static final String ROOT_LOCK_PATH = "shared_lock";
    
    private final CuratorFramework curatorFramework;
    
    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    
    // 根据序号排序
    private static Ordering<String> ordering = Ordering.from((Comparator<String>) (o1, o2) -> {
        
        String[] s1 = o1.split("-");
        String[] s2 = o2.split("-");
        
        return StringUtils.compare(s1[1], s2[1]);
    }).nullsLast();
    
    public SharedLocker(CuratorFramework curatorFramework) {
        this.curatorFramework = curatorFramework.usingNamespace("lock-namespace");
    }
    
    @Override
    public void afterPropertiesSet() {
        
        String path = "/" + ROOT_LOCK_PATH;
        try {
            if (curatorFramework.checkExists().forPath(path) == null) {
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(path);
            }
        } catch (Exception e) {
            LOGGER.error("[SharedLockByCurator afterPropertiesSet] connect zookeeper fail", e);
        }
    }
    
    @Override
    public String acquireDistributedLock(String path, String type) {
        
        String keyPath = "/" + ROOT_LOCK_PATH + "/" + path + "/" + type + "-";
        String parentPath = "/" + ROOT_LOCK_PATH + "/" + path;
        
        String thisPath = null;
        boolean haveTheLock = false;
        try {
            // 创建顺序临时节点
            thisPath = curatorFramework.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(keyPath);
            
            while (!haveTheLock) {
                
                // 获取子节点列表
                List<String> childPaths = curatorFramework.getChildren()
                        .forPath(parentPath);
                
                List<String> sortedPaths = ordering.sortedCopy(childPaths);
                // 获取当前节点的index位置
                int index = sortedPaths.indexOf(thisPath.split("/")[3]);
                
                if (index > 0) {
                    boolean flag = true;
                    // 读操作时 所有序号比该操作小的为读请求时才能获取锁
                    if (OPT_READ.equalsIgnoreCase(type)) {
                        for (int i = index - 1; i >= 0; i--) {
                            String temp = sortedPaths.get(i);
                            String tempType = temp.split("-")[0];
                            if (OPT_WRITE.equalsIgnoreCase(tempType)) {
                                addWatcher(parentPath + "/" + sortedPaths.get(i));
                                flag = false;
                                break;
                            }
                        }
                        
                    } else if (OPT_WRITE.equalsIgnoreCase(type)) {
                        // 写锁，检测有无写锁
                        for (int i = index - 1; i >= 0; i--) {
                            String temp = sortedPaths.get(i);
                            String tempType = temp.split("-")[0];
                            if (OPT_WRITE.equalsIgnoreCase(tempType)) {
                                addWatcher(parentPath + "/" + sortedPaths.get(i));
                                flag = false;
                                break;
                            }
                        }
                    }
                    
                    // 没有检测比自己小的写锁， 则获取有效的锁
                    if (flag) {
                        LOGGER.info("[SharedLockByCurator acquireDistributedLock] success thread= {} path= {} index= {}", Thread.currentThread().getName(), thisPath, index);
                        haveTheLock = true;
                    }
                    
                    if (countDownLatch.getCount() <= 0) {
                        countDownLatch = new CountDownLatch(1);
                    }
                    if (!haveTheLock) {
                        countDownLatch.await();
                    }
                    
                } else {
                    LOGGER.info("[SharedLockByCurator acquireDistributedLock] success thread= {} path= {} index= {}", Thread.currentThread().getName(), thisPath, index);
//                    addWatcher(thisPath);
                    haveTheLock = true;
                }
                
            }
            
        } catch (Exception e) {
            LOGGER.error("[SharedLockByCurator acquireDistributedLock] error", e);
        }
        
        return haveTheLock ? thisPath : "false";
    }
    
    @Override
    public String acquireDistributedLockNoWait(String path, String type) {
        return null;
    }
    
    private List<String> list = Collections.synchronizedList(Lists.newArrayList());
    
    private void addWatcher(String path) throws Exception {
        try {
            //  仅注册一次watcher
            if (curatorFramework.checkExists().forPath(path) != null && !list.contains(path)) {
                final NodeCache cache = new NodeCache(curatorFramework, path, false);
                cache.start(true);
                cache.getListenable().addListener(() -> {
                    LOGGER.info("[SharedLockByCurator nodeChanged] thread= {} path: {}", Thread.currentThread().getName(), path);
                    countDownLatch.countDown();
                    
                });
                list.add(path);
                LOGGER.info("[SharedLockByCurator addWatcher] thread= {} success path= {} ", Thread.currentThread().getName(), path);
            }
        } catch (Exception e) {
            LOGGER.error("[SharedLockByCurator addWatcher] error! ", e);
        }
    }
    
    @Override
    public boolean releaseDistributedLock(String keyPath) {
        try {
            if (curatorFramework.checkExists().forPath(keyPath) != null) {
                curatorFramework.delete().forPath(keyPath);
                LOGGER.info("[SharedLockByCurator releaseDistributedLock] thread= {} success to release lock for path: {}", Thread.currentThread().getName(), keyPath);
            }
        } catch (Exception e) {
            LOGGER.error("[SharedLockByCurator releaseDistributedLock] failed to release lock", e);
            return false;
        }
        return true;
    }
}
